package com.lhc.kafkademo.admin;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * Kafka 客户端操作
 */
public class AdminSample {

    public final static String TOPIC_NAME="jz_topic";

    public static void main(String[] args) throws Exception {

//        //获得AdminClient
//        AdminClient adminClient = AdminSample.adminClient();
//        System.out.println(adminClient.hashCode());

//        //创建topic
//        createTopic();

//        //获取topic 列表
//        topicLists();

//        //删除topic
//        delTopics();

//        //获取topic描述信息
//        describeTopics();

//        //获取config描述信息
//        describeConfig();
//        alterConfig();
//        describeConfig();

//        alterConfigNeo();
//        describeConfig();

        incrPartitions(2);
        describeTopics();


    }

    /**
     * 增加partitions 数量
     * @param partitions
     */
    public static void incrPartitions(int partitions) throws ExecutionException, InterruptedException {

        AdminClient adminClient = AdminSample.adminClient();

        Map<String, NewPartitions> partitionsMap = new HashMap<>();
        NewPartitions newPartitions = NewPartitions.increaseTo(partitions);
        partitionsMap.put(TOPIC_NAME, newPartitions);

        CreatePartitionsResult createPartitionsResult = adminClient.createPartitions(partitionsMap);
        createPartitionsResult.all().get();
    }


    /**
     * 修改config信息，新API
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void alterConfigNeo() throws ExecutionException, InterruptedException {
        AdminClient adminClient = adminClient();

        Map<ConfigResource,Collection<AlterConfigOp>> configMaps = new HashMap<>();
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);

        AlterConfigOp alterConfigOp = new AlterConfigOp(new ConfigEntry("preallocate", "false"), AlterConfigOp.OpType.SET);
        configMaps.put(configResource, Arrays.asList(alterConfigOp));

        AlterConfigsResult alterConfigsResult = adminClient.incrementalAlterConfigs(configMaps);
        alterConfigsResult.all().get();
    }

    /**
     * 修改config信息
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void alterConfig() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSample.adminClient();

        Map<ConfigResource,Config> configMaps = new HashMap<>();

        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        Config config = new Config(Arrays.asList(new ConfigEntry("preallocate", "true")));
        configMaps.put(configResource, config);

        AlterConfigsResult alterConfigsResult = adminClient.alterConfigs(configMaps);
        alterConfigsResult.all().get();
    }


    /**
     * 获取config描述信息
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void describeConfig() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSample.adminClient();

        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(configResource));
        Map<ConfigResource, Config> configResourceConfigMap = describeConfigsResult.all().get();
        Set<Map.Entry<ConfigResource, Config>> entries = configResourceConfigMap.entrySet();
        entries.stream().forEach((entry)->{
            System.out.println("configResource:"+entry.getKey()+";value:"+entry.getValue());
        });
    }



    /**
     * 获取topic描述信息
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void describeTopics() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSample.adminClient();
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(TOPIC_NAME));
        Map<String, TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get();
        Set<Map.Entry<String, TopicDescription>> entries = stringTopicDescriptionMap.entrySet();
        entries.stream().forEach((entry)->{
            System.out.println("name:"+entry.getKey()+";desc:"+entry.getValue());
        });
    }



    /**
     * 删除topic
     */
    public static void delTopics() throws ExecutionException, InterruptedException {
        AdminClient adminClient = AdminSample.adminClient();
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(TOPIC_NAME));
        deleteTopicsResult.all().get();
        System.out.println("delete:"+deleteTopicsResult);
    }

    /**
     * 列出所有topic
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public  static void topicLists() throws ExecutionException, InterruptedException {

        AdminClient adminClient = adminClient();
        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
        //是否查看internal选项
        listTopicsOptions.listInternal(true);
//        ListTopicsResult listTopicsResult = adminClient.listTopics();
        ListTopicsResult listTopicsResult = adminClient.listTopics(listTopicsOptions);  //列出__consumer_offsets
        Set<String> topics = listTopicsResult.names().get();
        //打印name
        topics.stream().forEach(System.out::println);

        Collection<TopicListing> topicListings = listTopicsResult.listings().get();
        //打印topicListings
        topicListings.stream().forEach((topicListing)->{
            System.out.println(topicListing);
        });
    }

    /**
     * 创建topic
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void createTopic() throws ExecutionException, InterruptedException {
        AdminClient adminClient = adminClient();

        Short rs = 1; // 副本因子
        NewTopic newTopic = new NewTopic(TOPIC_NAME,1, rs);
        CreateTopicsResult topics = adminClient.createTopics(Arrays.asList(newTopic));
        //等待创建，创建失败会报错
        topics.all().get();
        System.out.println("CreateTopicResult:"+topics);
    }

    /**
     * 获取客户端
     * @return
     */
    public static AdminClient adminClient(){

        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.11.130:9092");

        AdminClient adminClient = AdminClient.create(properties);

        return adminClient;
    }

}
